package org.infodavid.common.messaging.impl;

import java.io.InputStream;
import java.io.Serializable;
import java.util.Date;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.NamingException;

import net.sf.ehcache.Cache;
import net.sf.ehcache.CacheManager;
import net.sf.ehcache.Ehcache;
import net.sf.ehcache.Element;
import net.sf.ehcache.config.CacheConfiguration;
import net.sf.ehcache.event.CacheEventListener;
import net.sf.ehcache.store.MemoryStoreEvictionPolicy;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.infodavid.common.messaging.Constants;
import org.infodavid.common.messaging.MessagingService;
import org.infodavid.common.messaging.exceptions.MessagingException;

/**
 * The Class MessagingServiceImpl.
 */
public final class MessagingServiceImpl extends MessagingService {

  /**
   * The Class CacheEventListenerImpl.
   */
  private class CacheEventListenerImpl implements CacheEventListener {
    /*
     * (non-Javadoc)
     * @see java.lang.Object#clone()
     */
    @Override
    public Object clone() throws CloneNotSupportedException {
      return super.clone();
    }

    /**
     * Close the connection associated to the given element.
     * @param arg1 the element of the cache
     */
    private void close(final Element arg1) {
      final Object object = arg1.getObjectValue();

      if (object instanceof ConnectionEntry) {
        ((ConnectionEntry)object).close();
      }
    }

    /*
     * (non-Javadoc)
     * @see net.sf.ehcache.event.CacheEventListener#dispose()
     */
    public void dispose() {
      closeConnections();

      if (LOG.isDebugEnabled()) {
        LOG.debug("Cache disposed");
      }
    }

    /*
     * (non-Javadoc)
     * @see net.sf.ehcache.event.CacheEventListener#notifyElementEvicted(net.sf.ehcache .Ehcache,
     * net.sf.ehcache.Element)
     */
    public void notifyElementEvicted(final Ehcache arg0, final Element arg1) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Cache has evicted: " + arg1.getKey());
      }

      close(arg1);
    }

    /*
     * (non-Javadoc)
     * @see net.sf.ehcache.event.CacheEventListener#notifyElementExpired(net.sf.ehcache .Ehcache,
     * net.sf.ehcache.Element)
     */
    public void notifyElementExpired(final Ehcache arg0, final Element arg1) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Cache has detected expired: " + arg1.getKey());
      }

      close(arg1);
    }

    /*
     * (non-Javadoc)
     * @see net.sf.ehcache.event.CacheEventListener#notifyElementPut(net.sf.ehcache .Ehcache,
     * net.sf.ehcache.Element)
     */
    public void notifyElementPut(final Ehcache arg0, final Element arg1) {}

    /*
     * (non-Javadoc)
     * @see net.sf.ehcache.event.CacheEventListener#notifyElementRemoved(net.sf.ehcache .Ehcache,
     * net.sf.ehcache.Element)
     */
    public void notifyElementRemoved(final Ehcache arg0, final Element arg1) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Cache has removed: " + arg1.getKey());
      }

      close(arg1);
    }

    /*
     * (non-Javadoc)
     * @see net.sf.ehcache.event.CacheEventListener#notifyElementUpdated(net.sf.ehcache .Ehcache,
     * net.sf.ehcache.Element)
     */
    public void notifyElementUpdated(final Ehcache arg0, final Element arg1) {}

    /*
     * (non-Javadoc)
     * @see net.sf.ehcache.event.CacheEventListener#notifyRemoveAll(net.sf.ehcache .Ehcache)
     */
    public void notifyRemoveAll(final Ehcache arg0) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Cache removes all entries");
      }
    }
  }

  /**
   * The Class MessageListenerAdapter.
   */
  private static class MessageListenerAdapter implements MessageListener {

    /** The Constant LOG. */
    private static final Log LSNR_LOG = LogFactory.getLog(MessageListenerAdapter.class);

    /** The delegate listener. */
    private final transient IListener delegate;

    /**
     * Instantiates a new message listener adapter.
     * @param value the value
     */
    public MessageListenerAdapter(final IListener value) {
      if (value == null) {
        throw new IllegalArgumentException();
      }

      delegate = value;
    }

    /**
     * Returns the delegate.
     * @return the delegate
     */
    public IListener getDelegate() {
      return delegate;
    }

    /*
     * (non-Javadoc)
     * @see javax.jms.MessageListener#onMessage(javax.jms.Message)
     */
    public void onMessage(final Message message) {
      final Date now = new Date(System.currentTimeMillis());

      try {
        if (message instanceof TextMessage) {
          LSNR_LOG
              .debug("Text message recieved at " + now + " (" + message.getJMSMessageID() + ')');

          final TextMessage msg = (TextMessage)message;

          if (getDelegate().onTextMessage(msg.getText(), msg.getJMSTimestamp(),
              msg.getJMSMessageID())) {
            msg.acknowledge();
          }
        }
        else if (message == null) {
          LSNR_LOG.warn("Empty message recieved at " + now);
        }
        else {
          LSNR_LOG.debug("Message recieved at " + now + " (" + message.getJMSMessageID() + ')');

          Object value;

          if (message instanceof ObjectMessage) {
            value = ((ObjectMessage)message).getObject();
          }
          else {
            value = message;
          }

          if (getDelegate().onObjectMessage(value, message.getJMSTimestamp(),
              message.getJMSMessageID())) {
            message.acknowledge();
          }
        }
      }
      catch (final Exception e) {
        LSNR_LOG.error("An error occurs while retrieving a message", e);
      }
    }
  }

  /**
   * The Class QueueConnectionAdapter.
   */
  private class QueueConnectionAdapter implements IConnection {
    /** The session. */
    private transient Session session = null;
    /** The reciever. */
    private transient MessageConsumer reciever = null;
    /** The listener. */
    private transient MessageListener listener = null;
    /** The address. */
    private final transient String address;
    /** The destination name. */
    private final transient String name;

    /**
     * Constructor.
     * @param pAddress The address of the remote queue
     * @param pName The destination name
     * @param pListener The JMS listener to use
     * @throws JMSException the JMS exception
     * @throws NamingException the naming exception
     */
    public QueueConnectionAdapter(final String pAddress, final String pName,
        final MessageListener pListener) throws JMSException, NamingException {
      super();

      if (StringUtils.isEmpty(pAddress)) {
        throw new IllegalArgumentException("Address can not be null or empty");
      }
      else if (StringUtils.isEmpty(pName)) {
        throw new IllegalArgumentException("Name can not be null or empty");
      }
      else if (pListener == null) {
        throw new IllegalArgumentException("Listener can not be null");
      }
      else {
        address = pAddress;
        name = pName;
        listener = pListener;
      }
    }

    /**
     * Sets and links the listener to session and connection.
     * @param value The listener to link
     * @throws JMSException the JMS exception
     */
    public synchronized void setListener(final IListener value) throws JMSException {
      LOG.debug("Setting listener");

      listener = new MessageListenerAdapter(value);
    }

    public synchronized void connect() throws JMSException, NamingException, MessagingException {
      LOG.debug("Starting listening");

      final ConnectionEntry entry = getConnectionEntry(address);
      final Connection connection = entry.getConnection();

      if (getOptions().isAutoAcknowledge()) {
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      }
      else {
        session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
      }

      reciever = session.createConsumer(entry.getDestination(name));

      reciever.setMessageListener(listener);
      connection.start();
    }

    /*
     * (non-Javadoc)
     * @see org.infodavid.common.messaging.MessagingService.IConnection#disconnect ()
     */
    public void disconnect() throws Exception {
      if (reciever != null) {
        LOG.debug("Stopping listening");

        reciever.setMessageListener(null);

        LOG.debug("Closing reciever.");

        try {
          reciever.close();
        }
        catch (final JMSException e) {
          LOG.debug("Error when closing reciever: ", e);
        }
      }

      if (session != null) {
        LOG.debug("Closing session.");

        try {
          if (session.getTransacted()) {
            session.commit();
          }
        }
        catch (final JMSException e) {
          LOG.debug("Error while closing session: ", e);
        }

        try {
          session.close();
        }
        catch (final Exception e) {
          LOG.debug("An error occurs while disconnecting", e);
        }
      }
    }
  }

  /** The Constant MESSAGING_SERVICE_CACHE. */
  private static final String MESSAGING_SERVICE_CACHE = "MessagingServiceCache";

  /** The Constant CONFIG_FILE. */
  protected static final String CONFIG_FILE = "messaging_cache.xml";

  /** The Constant TIME_TO_IDLE_SYSTEM_PROPERTY. */
  public static final String TIME_TO_IDLE_SYSTEM_PROPERTY;

  /** The Constant TIME_TO_LIVE_SYSTEM_PROPERTY. */
  public static final String TIME_TO_LIVE_SYSTEM_PROPERTY;

  /** The Constant MAX_ELEMENTS_SYSTEM_PROPERTY. */
  public static final String MAX_ELEMENTS_SYSTEM_PROPERTY;

  /** The Constant LOG. */
  private static final Log LOG = LogFactory.getLog(MessagingService.class);

  /** The Constant DEFAULT_TIME_TO_LIVE. */
  public static final int DEFAULT_TIME_TO_LIVE = 240;

  /** The Constant DEFAULT_TIME_TO_IDLE. */
  public static final int DEFAULT_TIME_TO_IDLE = 300;

  /** The Constant DEFAULT_MAX_ELEMENTS. */
  public static final int DEFAULT_MAX_ELEMENTS = 20;

  static {
    final String packageName = MessagingService.class.getPackage().getName();

    TIME_TO_IDLE_SYSTEM_PROPERTY = packageName + ".timeToIdleSeconds";
    TIME_TO_LIVE_SYSTEM_PROPERTY = packageName + ".timeToLiveSeconds";
    MAX_ELEMENTS_SYSTEM_PROPERTY = packageName + ".maxElements";
  }

  /** The cache of connection entries. */
  private final Cache cache;

  /**
   * Constructor.
   */
  private MessagingServiceImpl() {
    LOG.debug("Building MessagingService");

    cache = buildCache();

    Runtime.getRuntime().addShutdownHook(new Thread() {
      /*
       * (non-Javadoc)
       * @see java.lang.Thread#run()
       */
      @Override
      public void run() {
        cache.dispose();
      }
    });
  }

  /**
   * Build the cache according to properties described in the properties file and to system
   * properties.
   * @return The initialized cache object
   */
  private Cache buildCache() {
    InputStream in = null;
    CacheManager manager = null;
    Cache lCache = null;

    LOG.debug("Loading " + MESSAGING_SERVICE_CACHE + " from configuration file " + CONFIG_FILE);

    try {
      in = getClass().getClassLoader().getResourceAsStream(CONFIG_FILE);
      manager = new CacheManager(in);

      lCache =
          new Cache(MESSAGING_SERVICE_CACHE, DEFAULT_MAX_ELEMENTS, MemoryStoreEvictionPolicy.LFU,
              false, System.getProperty("java.io.tmp"), false, DEFAULT_TIME_TO_LIVE,
              DEFAULT_TIME_TO_IDLE, false, Cache.DEFAULT_EXPIRY_THREAD_INTERVAL_SECONDS, null);
      lCache.getCacheEventNotificationService().registerListener(new CacheEventListenerImpl());

      // Setup of some cache properties according to system properties
      final CacheConfiguration config = lCache.getCacheConfiguration();
      String value = System.getProperty(TIME_TO_IDLE_SYSTEM_PROPERTY);

      if (StringUtils.isNotEmpty(value)) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Setting value " + value + " for property: " + TIME_TO_IDLE_SYSTEM_PROPERTY);
        }

        try {
          config.setTimeToIdleSeconds(Long.parseLong(value));
        }
        catch (final Exception e) {
          LOG.warn("Invalid value for property: " + TIME_TO_IDLE_SYSTEM_PROPERTY, e);
        }
      }

      value = System.getProperty(TIME_TO_LIVE_SYSTEM_PROPERTY);

      if (StringUtils.isNotEmpty(value)) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Setting value " + value + " for property: " + TIME_TO_LIVE_SYSTEM_PROPERTY);
        }

        try {
          config.setTimeToLiveSeconds(Long.parseLong(value));

          if (config.getTimeToLiveSeconds() <= 0) {
            if (LOG.isDebugEnabled()) {
              LOG.debug("No time to live, eternal entries for messaging connections");
            }

            config.setEternal(true);
          }
        }
        catch (final Exception e) {
          LOG.warn("Invalid value for property: " + TIME_TO_LIVE_SYSTEM_PROPERTY, e);
        }
      }

      value = System.getProperty(MAX_ELEMENTS_SYSTEM_PROPERTY);

      if (StringUtils.isNotEmpty(value)) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Setting value " + value + " for property: " + MAX_ELEMENTS_SYSTEM_PROPERTY);
        }

        try {
          config.setMaxElementsInMemory(Integer.parseInt(value));
        }
        catch (final Exception e) {
          LOG.warn("Invalid value for property: " + MAX_ELEMENTS_SYSTEM_PROPERTY, e);
        }
      }

      manager.addCache(lCache);
    }
    finally {
      if (in != null) {
        try {
          in.close();
        }
        catch (final Exception e) {
          LOG.debug("Failed to close stream on configuration file: " + CONFIG_FILE, e);
        }
      }
    }

    return lCache;
  }

  /*
   * (non-Javadoc)
   * @see org.infodavid.common.messaging.MessagingService#closeConnections()
   */
  @Override
  public void closeConnections() {
    LOG.debug("Begin closeConnections()");

    synchronized (cache) {
      for (final Object key : cache.getKeys()) {
        cache.remove(key);
      }
    }

    LOG.debug("End closeConnections()");
  }

  /**
   * See super class or interface. (non-Javadoc)
   * @see org.infodavid.common.messaging.MessagingService#sendObjectMessages(java.lang.String,
   *      java.io.Serializable[])
   */
  @Override
  public void sendObjectMessages(final String queueName, final Serializable... objects)
      throws MessagingException {
    sendObjectMessages(null, queueName, objects);
  }

  /**
   * See super class or interface. (non-Javadoc)
   * @see org.infodavid.common.messaging.MessagingService#sendObjectMessages(java.lang.String,
   *      java.lang.String, org.infodavid.common.messaging.MessagingService.MessagingOptions,
   *      java.io.Serializable[])
   */
  @Override
  public void sendObjectMessages(final String address, final String queueName,
      final MessagingOptions pOptions, final Serializable... objects) throws MessagingException {
    sendMessages(address, queueName, pOptions, objects);
  }

  /**
   * See super class or interface. (non-Javadoc)
   * @see org.infodavid.common.messaging.MessagingService#sendObjectMessages(java.lang.String,
   *      org.infodavid.common.messaging.MessagingService.MessagingOptions, java.io.Serializable[])
   */
  @Override
  public void sendObjectMessages(final String queueName, final MessagingOptions pOptions,
      final Serializable... objects) throws MessagingException {
    sendMessages(null, queueName, pOptions, objects);
  }

  /**
   * See super class or interface. (non-Javadoc)
   * @see org.infodavid.common.messaging.MessagingService#sendTextMessages(java.lang.String,
   *      java.lang.String, java.lang.String[])
   */
  @Override
  public void sendTextMessages(final String address, final String queueName, final String... texts)
      throws MessagingException {
    sendMessages(address, queueName, getOptions(), texts);
  }

  /**
   * See super class or interface. (non-Javadoc)
   * @see org.infodavid.common.messaging.MessagingService#sendTextMessages(java.lang.String,
   *      java.lang.String, org.infodavid.common.messaging.MessagingService.MessagingOptions,
   *      java.lang.String[])
   */
  @Override
  public void sendTextMessages(final String address, final String queueName,
      final MessagingOptions pOptions, final String... texts) throws MessagingException {
    sendMessages(address, queueName, pOptions, texts);
  }

  /**
   * See super class or interface. (non-Javadoc)
   * @see org.infodavid.common.messaging.MessagingService#sendTextMessages(java.lang.String,
   *      org.infodavid.common.messaging.MessagingService.MessagingOptions, java.lang.String[])
   */
  @Override
  public void sendTextMessages(final String queueName, final MessagingOptions pOptions,
      final String... texts) throws MessagingException {
    sendMessages(null, queueName, pOptions, texts);
  }

  /**
   * See super class or interface. (non-Javadoc)
   * @see org.infodavid.common.messaging.MessagingService#sendObjectMessages(java.lang.String,
   *      java.lang.String, java.io.Serializable[])
   */
  @Override
  public void sendObjectMessages(final String address, final String queueName,
      final Serializable... objects) throws MessagingException {
    sendMessages(address, queueName, getOptions(), objects);
  }

  /**
   * See super class or interface. (non-Javadoc)
   * @see org.infodavid.common.messaging.MessagingService#sendTextMessages(java.lang.String,
   *      java.lang.String[])
   */
  @Override
  public void sendTextMessages(final String queueName, final String... texts)
      throws MessagingException {
    sendMessages(null, queueName, getOptions(), texts);
  }

  /*
   * (non-Javadoc)
   * @see org.infodavid.common.messaging.MessagingService#listen(java.lang.String, java.lang.String,
   * org.infodavid.common.messaging.MessagingService.IListener)
   */
  @Override
  public IConnection listen(final String address, final String queueName, final IListener listener)
      throws MessagingException {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Begin listen using address = " + address + ", queueName = " + queueName
          + ", listener = " + listener);
    }

    QueueConnectionAdapter connection = null;

    if (StringUtils.isNotEmpty(queueName)) {
      throw new IllegalArgumentException("Trying to listen to messages without queue name");
    }
    else if (listener == null) {
      throw new IllegalArgumentException("Trying to listen to messages without listener");
    }
    else {
      MessageListenerAdapter adapter;

      LOG.debug("Creating MessageListenerAdapter.");

      try {
        adapter = new MessageListenerAdapter(listener);

        LOG.debug("Creating QueueConnection adapter.");

        connection = new QueueConnectionAdapter(address, queueName, adapter);
      }
      catch (final Exception e) {
        throw new MessagingException("An error occurs while listening to messages", e);
      }
    }

    LOG.debug("End listen");

    return connection;
  }

  /**
   * Gets the client retry count.
   * @return the retry count
   */
  private int getRetryCount() {
    final String value = System.getProperty(Constants.CLIENT_RETRY_COUNT_SYSTEM_PROPERTY);
    int result = Constants.DEFAULT_CLIENT_RETRY_COUNT;

    if (StringUtils.isNotEmpty(value)) {
      try {
        result = Double.valueOf(value).intValue();
      }
      catch (final Exception e) {
        result = -1;
      }
    }

    if (result < 0) {
      result = Constants.DEFAULT_CLIENT_RETRY_COUNT;
    }

    return result;
  }

  /**
   * Gets the client retry period.
   * @return the retry period
   */
  private long getRetryPeriod() {
    final String value = System.getProperty(Constants.CLIENT_RETRY_PERIOD_SYSTEM_PROPERTY);
    long result = Constants.DEFAULT_CLIENT_RETRY_PERIOD;

    if (StringUtils.isNotEmpty(value)) {
      try {
        result = Double.valueOf(value).longValue();
      }
      catch (final Exception e) {
        result = -1;
      }
    }

    if (result < 0) {
      result = Constants.DEFAULT_CLIENT_RETRY_PERIOD;
    }

    return result;
  }

  /**
   * Send messages. TextMessage objects are instances of String or ObjectMessage objects if not
   * @param address the address
   * @param queueName the queue name
   * @param pOptions the options
   * @param objects the objects
   */
  protected void sendMessages(final String address, final String queueName,
      final MessagingOptions pOptions, final Serializable[] objects) throws MessagingException {
    final long retryPeriod = getRetryPeriod();
    final int retryCount = getRetryCount();
    int retries = 0;

    do {
      try {
        processMessages(address, queueName, pOptions, objects);
        retries = retryCount;
      }
      catch (final Throwable e) {
        if (retries >= retryCount) {
          if (e instanceof MessagingException) {
            throw (MessagingException)e;
          }

          throw new MessagingException("An error occurs while sending of the messages", e);
        }

        LOG.warn("An error occurs while sending of the messages (retry=" + retries
            + "), retrying after " + retryPeriod + " ms", e);

        try {
          Thread.sleep(retryPeriod);
        }
        catch (final InterruptedException e1) {
          LOG.warn("An error occurs while setting retry period", e1);
        }

        retries++;
      }
    }
    while (retries < retryCount);
  }

  /**
   * Send messages. TextMessage objects are instances of String or ObjectMessage objects if not
   * @param address the address
   * @param destinationName the destination name
   * @param pOptions the options
   * @param objects the objects
   */
  protected void processMessages(final String address, final String destinationName,
      final MessagingOptions pOptions, final Serializable[] objects) throws Exception {
    if (StringUtils.isEmpty(destinationName)) {
      throw new IllegalArgumentException("Trying to send a message without destination name");
    }
    else if (objects == null || objects.length == 0) {
      throw new IllegalArgumentException("Trying to send a message without objects to send");
    }
    else {
      Session session = null;
      MessageProducer sender = null;
      ConnectionEntry entry = null;
      Message message;
      Destination destination;
      final long timeToLive = pOptions.getTimeToLiveInMillis();
      final int priority = pOptions.getPriority();
      final boolean persistent = pOptions.isPersistentDelivery();
      final boolean autoAck = pOptions.isAutoAcknowledge();

      try {
        // TODO enhancement to use clustering and multiple addresses
        // multiple addresses must be extracted to test the connection on each
        // one until an error occurs
        entry = getConnectionEntry(address);

        LOG.debug("Creating session.");

        if (autoAck) {
          session = entry.getConnection().createSession(false, Session.AUTO_ACKNOWLEDGE);
        }
        else {
          session = entry.getConnection().createSession(false, Session.CLIENT_ACKNOWLEDGE);
        }

        destination = entry.getDestination(destinationName);

        LOG.debug("Creating sender.");

        sender = session.createProducer(destination);

        /*
         * JBoss Messaging does not support disabling message ID generation
         * sender.setDisableMessageID(false); sender.setDisableMessageTimestamp(false);
         */

        if (timeToLive > 0) {
          sender.setTimeToLive(timeToLive);
        }
        else {
          sender.setTimeToLive(Message.DEFAULT_TIME_TO_LIVE);
        }

        if (persistent) {
          sender.setDeliveryMode(DeliveryMode.PERSISTENT);
        }
        else {
          sender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        }

        if (priority >= 0 && priority <= 9) {
          sender.setPriority(priority);
        }
        else {
          sender.setPriority(Message.DEFAULT_PRIORITY);
        }

        for (final Serializable obj : objects) {
          if (obj instanceof String) {
            LOG.debug("Creating TextMessage.");

            message = session.createTextMessage((String)obj);
          }
          else {
            LOG.debug("Creating ObjectMessage.");

            message = session.createObjectMessage(obj);
          }

          if (LOG.isDebugEnabled()) {
            LOG.debug("Sending message to destination: " + destinationName);
          }

          sender.send(message);

          LOG.info("Message sent to destination: " + destinationName + " successfully.");
        }
      }
      finally {
        if (sender != null) {
          LOG.debug("Closing sender.");

          try {
            sender.close();
          }
          catch (final JMSException e) {
            LOG.error("Error when closing sender: ", e);
          }
          catch (final Throwable e) {
            LOG.error("Unexpected exception when closing sender: ", e);
          }
        }

        if (session != null) {
          LOG.debug("Closing session.");

          try {
            if (session.getTransacted()) {
              session.commit();
            }
          }
          catch (final JMSException e) {
            LOG.warn("Error when closing session: ", e);
          }

          try {
            session.close();
          }
          catch (final JMSException e) {
            LOG.warn("Error when closing session: ", e);
          }
        }
      }
    }
  }

  /**
   * The description of the object.
   * @return the description of the object.
   */
  @Override
  public String toString() {
    final StringBuffer buffer = new StringBuffer();

    buffer.append(super.toString());
    buffer.append('[');
    buffer.append(" cache = ").append(cache);
    buffer.append(" options = ").append(getOptions());
    buffer.append(']');

    return buffer.toString();
  }

  /**
   * Free resources.
   * @throws Throwable the throwable
   */
  @Override
  protected void finalize() throws Throwable {
    try {
      closeConnections();
    }
    catch (final Throwable e) {
      LOG.warn("An error occurs when closing JMS connections", e);
    }

    try {
      CacheManager.getInstance().shutdown();
    }
    catch (final Throwable e) {
      LOG.debug("An error occurs while shutdown of cache manager.", e);
    }

    super.finalize();
  }

  /**
   * Gets the connection entry from the cache or build and register the new one.
   * @param address the address
   * @return the connection entry
   */
  protected ConnectionEntry getConnectionEntry(final String address) {
    LOG.debug("Begin getConnectionEntry");

    ConnectionEntry object = null;

    if (cache == null) {
      throw new IllegalAccessError("Cache has not been initialized");
    }

    Element element;

    synchronized (cache) {
      element = cache.get(address);

      if (element == null) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Connection entry not found for the given key: " + address);
          LOG.debug("Inserting a new Connection entry into cache");
        }

        object = new ConnectionEntry(address);
        element = new Element(address, object);
        cache.put(element);
      }
      else {
        object = (ConnectionEntry)element.getObjectValue();
        // not needed, done when requesting cache to retrieve the element:
        // element.updateUpdateStatistics();
      }

      if (LOG.isDebugEnabled()) {
        LOG.debug("Number of times requested items were found in the cache: "
            + cache.getStatistics().getCacheHits());
      }
    }

    return object;
  }
}
